package uk.org.mygrid.eobesity.utils;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.logging.Level;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.Ostermiller.util.CSVParser;
import com.Ostermiller.util.CSVPrinter;
import com.Ostermiller.util.LabeledCSVParser;

public class CSVUtils {

	private static Logger logger = Logger.getLogger(CSVUtils.class);
	private static Properties mailServerConfig = new Properties();
	private Configuration config;

	public CSVUtils() {
	}

	/**
	 * 
	 * @param files
	 *            the list of files to add to the zip
	 * @param fileName
	 *            the jobID UUID
	 * @return
	 */
	public String zipWriter(List<File> files, String fileName) {

		byte[] buf = new byte[1024];

		// String tempDirectory = Messages.getString("DownloadFile.31");
		String tempDirectory = getConfig().getString("DownloadFile.31");
		String outFilename = null;

		try {
			// Create the ZIP file

			outFilename = tempDirectory + fileName + ".zip"; //$NON-NLS-1$ //$NON-NLS-2$
			ZipOutputStream out = new ZipOutputStream(new FileOutputStream(
					outFilename));

			// Compress the files

			for (File file : files) {
				FileInputStream in = new FileInputStream(file); //$NON-NLS-1$

				// remove the .UUID part and replace with '_selection.csv'
				// String name = file.getName().replaceAll("." + fileName,
				// "_selection.csv");

				// Add ZIP entry to output stream.
				out.putNextEntry(new ZipEntry(file.getName())); //$NON-NLS-1$

				// Transfer bytes from the file to the ZIP file
				int len;
				while ((len = in.read(buf)) > 0) {
					out.write(buf, 0, len);
				}

				// Complete the entry
				out.closeEntry();
				in.close();
			}

			// Complete the ZIP file
			out.close();
			// remove the old files since they are now inside the zip
			for (File file : files) {
				// FileUtils.forceDelete(file);
			}
			logger.info("Created ZIP file for " + fileName);
		} catch (IOException e) {
			logger.error("Problem creating ZIP file " + fileName + ": " + e);
		}
		return outFilename;
	}

	/**
	 * Given a dataset name and a Map of the variables to their values write out
	 * a CSV file with these values in it
	 * 
	 * @param dataset
	 *            the name os a dataset
	 * @param valueMap
	 *            the values for each of the variables
	 * @return the CSV file written out to disk
	 */
	public void CSVWriter(List<String> values, boolean header,
			CSVPrinter csvPrinter) {
		// csvPrinter.changeDelimiter('\t');
		//write the column headers
		if (header) {
			// get the names of the columns and write to the CSV
			String[] headerNames = new String[values.size()];
			values.toArray(headerNames);
			try {
				// csvPrinter.writeln(varKeys);
				csvPrinter.writeln(headerNames);
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		}
		//write the actual values
		if (!header) {
			String[] valueLineArray = new String[values.size()];
			values.toArray(valueLineArray);
			try {
				csvPrinter.writeln(valueLineArray);
			} catch (IOException e) {
				logger.error("Could not write line to csv file ", e);
			}
		}

	}

	/**
	 * Uses {@link LabeledCSVParser} to parse a CSV type file which has the
	 * column names at the top
	 * 
	 * @param file
	 * @return
	 * @throws FileNotFoundException
	 */
	public File readFieldNames(String dataset,
			String survey, String year, String actualDatasetName, List<String> variables)
			throws FileNotFoundException {
		// Map<String, List<String>> varMap = new HashMap<String,
		// List<String>>();
		// String directory = Messages.getString("RTFSearcher.0");
		String directory = getConfig().getString("RTFSearcher.0");
		// stop people from requesting any file
		File rootDirectory = new File(directory);
		String[] list = rootDirectory.list();
		List<String> asList2 = Arrays.asList(list);

		
		if (!asList2.contains(dataset)) {
			throw new FileNotFoundException("Dataset " + dataset
					+ "does not exist in directory " + rootDirectory);
		}

		File file = new File(directory + dataset);

		BufferedInputStream bufferedInputStream = null;
		try {
			bufferedInputStream = new BufferedInputStream(new FileInputStream(
					file));
		} catch (FileNotFoundException e1) {
			logger.error(e1);
		}
		CSVParser parser = new CSVParser(bufferedInputStream, '\t');

		LabeledCSVParser labelParser = null;
		try {
			labelParser = new LabeledCSVParser(parser);
		} catch (IOException e) {
			e.printStackTrace();
		}
		String[] labels = null;
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		// writer
		String tempDirectory = getConfig().getString("DownloadFile.31");
		// tempDirectory = tempDirectory + File.pathSeparator +
		// UUID.randomUUID().toString() + File.pathSeparator;
		File topLevel = new File(tempDirectory, UUID.randomUUID().toString());
		try {
			FileUtils.forceMkdir(topLevel);
		} catch (IOException ex) {
			logger.error("Could not make directory " + topLevel);
		}
		File outputfile = null;

		outputfile = new File(topLevel,
				actualDatasetName.split("\\.")[0] + "_selection.csv"); //$NON-NLS-1$
		FileOutputStream outputStream = null;
		try {
			outputStream = new FileOutputStream(outputfile);
		} catch (FileNotFoundException e1) {
			logger.error("Problem writing  dataset file: " + e1);
		}
		CSVPrinter csvPrinter = new CSVPrinter(outputStream);
		List<String> newVariableList = new ArrayList<String>(variables);
		newVariableList.add(0, "row");
		newVariableList.add(1, "dataset");
		newVariableList.add(2, "survey");
		newVariableList.add(3, "year");
		CSVWriter(newVariableList, true, csvPrinter);
		List<String> asList = Arrays.asList(labels);
		String[] parseLine;
		try {
			//row starting at 1
			int i = 1;
			boolean first = true;
			//read and write lines one by one
			String datasetSplitName = actualDatasetName.split("\\.")[0];
			while ((parseLine = labelParser.getLine()) != null) {
				List<String> valueList = new ArrayList<String>();
				valueList.add(Integer.toString(i));
				valueList.add(datasetSplitName);
				valueList.add(survey);
				valueList.add(year);
				for (String varLabel : variables) {
					// check to see if the metadata case matches the header
					// first time round
					if (first) {
						if (asList.contains(varLabel.toLowerCase())) {
							variables.set(variables.indexOf(varLabel), varLabel
									.toLowerCase());
							varLabel = varLabel.toLowerCase();
						} else {
							variables.set(variables.indexOf(varLabel), varLabel
									.toUpperCase());
							varLabel = varLabel.toUpperCase();
						}
					}
					String value = labelParser.getValueByLabel(varLabel);
					if (value == null) {
						value = "X";
					}
					valueList.add(value);

				}
				// write out line here
				CSVWriter(valueList, false, csvPrinter);
				first = false;
				i++;
			}
		} catch (IOException e2) {
			logger.error("Problem reading line from CSV file " + dataset
					+ " with variables " + variables);

		}
		logger.info("Loaded variables from " + dataset);
		
		try {
			csvPrinter.flush();
			csvPrinter.close();

		} catch (IOException e) {
		}
		return outputfile;
	}

	/**
	 * Uses {@link HTable} to load columns for the dataset and populate a
	 * {@link Map} for each variable requested
	 * 
	 * @param file
	 * @return
	 * @throws FileNotFoundException
	 */
	public Map<String, List<String>> readFieldNamesUsingHBase(String dataset,
			List<String> variables) throws FileNotFoundException {

		Map<String, List<String>> varMap = new HashMap<String, List<String>>();

		HBaseConfiguration config = new HBaseConfiguration();
		HTable table = null;
		try {
			// TODO this database name should be a parameter when there are
			// multiple dataset types
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}
		// check if the variable label is upper or lower case
		for (String variable : variables) {
			// use the first row for convenience
			Get g = new Get(Bytes.toBytes(1));
			g.addColumn(Bytes.toBytes(dataset), Bytes.toBytes(variable
					.toLowerCase()));
			boolean colExists = false;

			try {
				colExists = table.exists(g);
			} catch (IOException ex) {
				logger.error(ex);
			}
			if (!colExists) {
				// set the label to be upper case
				logger.info("Changed case of " + variable);
				variables.set(variables.indexOf(variable), variable
						.toUpperCase());
			}

		}
		// we have the correct variable name case, now get the data
		Scan scan = new Scan();
		for (String variable : variables) {
			scan.addColumn(Bytes.toBytes(dataset), Bytes.toBytes(variable));
		}
		try {
			ResultScanner resultScanner = table.getScanner(scan);
			// get each row of data for each variable
			logger.info("Scanning HBase results for " + dataset);
			for (Result result : resultScanner) {
				int i = 0;
				for (String variable : variables) {
					List<String> valueList = null;
					try {
						valueList = varMap.get(variable);
						if (valueList == null) {
							valueList = new ArrayList<String>();
							varMap.put(variable, valueList);
						}
						byte[] value = result
								.getCellValue(Bytes.toBytes(dataset),
										Bytes.toBytes(variable)).getValue();

						valueList.add(Bytes.toString(value));
					} catch (Exception e) {
						// if a value is missing then put in 'X'
						logger
								.error("Problem with variable " + variable
										+ " ,row " + i
										+ ", adding default value as X ", e);
						valueList.add("X");
					}
					i++;
				}
			}
		} catch (IOException ex) {
			logger.error(ex);
		}
		logger.info("Loaded variables from " + dataset);
		return varMap;

	}

	public void extractVariablesFromDataset(String dataset) {
		Map<String, List<String>> varMap = new HashMap<String, List<String>>();
		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		try {
			labelParser.close();
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}
		List<String> asList = Arrays.asList(labels);
		// loop through each column in turn
		for (String label : asList) {
			System.out.println("Parsing label " + label);
			logger.info("Parsing label " + label);
			String[] parseLine = null;
			try {
				labelParser = loadCSVFile(dataset);
				// jump past the labels
				labelParser.getLine();

				List<String> valList = new ArrayList<String>();
				int i = 0;
				while ((parseLine = labelParser.getLine()) != null) {
					System.out.println("Parsing line " + i);
					logger.info("Parsing line " + i);
					valList.add(labelParser.getValueByLabel(label));

					i++;
				}
				labelParser.close();
			} catch (IOException ex) {
				logger.error("Could not parse line " + ex);
			}
		}

	}

	public void extractVariablesFromDatasetUsingSplit(String dataset) {
		// Map<String, List<String>> varMap = new HashMap<String,
		// List<String>>();
		BufferedReader reader = loadFile(dataset);
		String[] labels = null;
		try {
			labels = reader.readLine().split("\t");
			reader.close();
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}
		List<String> asList = Arrays.asList(labels);
		// loop through each column in turn
		for (String label : asList) {
			System.out.println("Parsing label " + label);
			logger.info("Parsing label " + label);
			String parseLine = null;
			try {
				reader = loadFile(dataset);
				// jump past the labels
				reader.readLine();

				List<String> valList = new ArrayList<String>();
				int i = 0;
				while ((parseLine = reader.readLine()) != null) {
					String[] values = parseLine.split("\t");
					System.out.println("Parsing line " + i);
					logger.info("Parsing line " + i);
					valList.add(values[asList.indexOf(label)]);

					i++;
				}
				reader.close();
			} catch (IOException ex) {
				logger.error("Could not parse line " + ex);
			}
		}

	}

	public void loadCSVIntoHBase(String dataset) {

		HBaseConfiguration config = new HBaseConfiguration();
		// get the table from HBase
		HTable table = null;
		try {
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}
		// retrieve row1
		// Get g = new Get(Bytes.toBytes(2));
		// Result r = null;
		// try {
		// r = table.get(g);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		//
		// byte[] value = r.getValue(Bytes.toBytes("test_column"),
		// Bytes.toBytes("col1"));
		//
		// System.out.println(new String(tableName));
		// System.out.println(Bytes.toInt(value));

		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		// get the column labels/headers
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		// try {
		// labelParser.close();
		// } catch (IOException ex) {
		// java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(Level.SEVERE,
		// null, ex);
		// }
		List<String> labelList = Arrays.asList(labels);
		try {
			// labelParser = loadCSVFile(dataset);
			// //jump past the labels
			// labelParser.getLine();

			// i is for the row value
			int i = 0;
			while (labelParser.getLine() != null) {
				// get each value from the row
				System.out.println("Parsing line " + i);
				logger.info("Parsing line " + i);
				for (String label : labelList) {
					// add the single csv value to table HSE, column family
					// HSE2006, column 'label', row 'i'
					Put p = new Put(Bytes.toBytes(i));
					p.add(Bytes.toBytes(dataset.split("\\.")[0]), Bytes
							.toBytes(label), Bytes.toBytes(labelParser
							.getValueByLabel(label)));
					try {
						table.put(p);
					} catch (IOException ex) {
						logger.error(ex);
						System.out.println(ex);
					}
				}
				i++;
			}
			System.out.println("There are " + i + "rows");
			table.close();
			labelParser.close();
		} catch (IOException ex) {
			logger.error("Could not parse line " + ex);
		}
	}

	public void loadCSVIntoHBaseColumnByColumn(String dataset) {

		HBaseConfiguration config = new HBaseConfiguration();
		// get the table from HBase
		HTable table = null;
		try {
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}
		// retrieve row1
		// Get g = new Get(Bytes.toBytes(2));
		// Result r = null;
		// try {
		// r = table.get(g);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		//
		// byte[] value = r.getValue(Bytes.toBytes("test_column"),
		// Bytes.toBytes("col1"));
		//
		// System.out.println(new String(tableName));
		// System.out.println(Bytes.toInt(value));

		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		// get the column labels/headers
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		try {
			labelParser.close();
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}
		List<String> labelList = Arrays.asList(labels);
		int i = 0;
		for (String label : labelList) {
			loadCSVColumnIntoHBase(dataset, label);
			logger.info(label + " : complete : " + i);
			i++;
		}
		try {
			table.close();
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}

	}

	public void loadCSVColumnIntoHBase(String dataset, String column) {

		HBaseConfiguration config = new HBaseConfiguration();
		// get the table from HBase
		HTable table = null;
		try {
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}
		// retrieve row1
		// Get g = new Get(Bytes.toBytes(2));
		// Result r = null;
		// try {
		// r = table.get(g);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		//
		// byte[] value = r.getValue(Bytes.toBytes("test_column"),
		// Bytes.toBytes("col1"));
		//
		// System.out.println(new String(tableName));
		// System.out.println(Bytes.toInt(value));

		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		// get the column labels/headers
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		// try {
		// labelParser.close();
		// } catch (IOException ex) {
		// java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(Level.SEVERE,
		// null, ex);
		// }
		List<String> labelList = Arrays.asList(labels);
		try {
			// labelParser = loadCSVFile(dataset);
			// //jump past the labels
			// labelParser.getLine();

			// i is for the row value
			int i = 0;
			while (labelParser.getLine() != null) {
				// get each value from the row
				// System.out.println("Parsing line " + i);
				// logger.info("Parsing line " + i);
				String value = labelParser.getValueByLabel(column);
				// add the single csv value to table HSE, column family HSE2006,
				// column 'label', row 'i'
				Put p = new Put(Bytes.toBytes(i));
				p.add(Bytes.toBytes(dataset.split("\\.")[0]), Bytes
						.toBytes(column), Bytes.toBytes(value));
				// logger.info(i + " : " + value);
				try {
					table.put(p);
				} catch (IOException ex) {
					logger.error(ex);
					System.out.println(ex);
				}

				i++;
			}
			logger.info(column + "has " + i + " rows");
			labelParser.close();
		} catch (IOException ex) {
			logger.error("Could not parse line " + ex);
		}
	}

	public void readDatasetTest(String dataset) {

		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		// get the column labels/headers
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		String firstLabel = labels[0];
		int i = 0;
		try {
			while (labelParser.getLine() != null) {
				logger
						.info(i + " : "
								+ labelParser.getValueByLabel(firstLabel));
				i++;
			}
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}
		logger.info(firstLabel);
	}

	public void testHBase(String columnFamily) {
		HBaseConfiguration config = new HBaseConfiguration();
		HTable table = null;
		try {
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}
		// byte[] tableName = table.getTableName();

		// Put p;
		// //row 1 col1
		// p = new Put(Bytes.toBytes(1));
		// p.add(Bytes.toBytes(""), Bytes.toBytes("col1"),
		// Bytes.toBytes(21));
		// try {
		// table.put(p);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		// //row 2 col1
		// p = new Put(Bytes.toBytes(2));
		// p.add(Bytes.toBytes("test_column"), Bytes.toBytes("col1"),
		// Bytes.toBytes(22));
		// try {
		// table.put(p);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		//
		// p = new Put(Bytes.toBytes(1));
		// //row1 col2
		// p.add(Bytes.toBytes("test_column"), Bytes.toBytes("col2"),
		// Bytes.toBytes(23));
		//
		// try {
		// table.put(p);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }

		Get g = new Get(Bytes.toBytes(1));
		g.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("finoutc"));
		boolean colExists = false;
		try {
			colExists = table.exists(g);
		} catch (IOException ex) {
			java.util.logging.Logger.getLogger(CSVUtils.class.getName()).log(
					Level.SEVERE, null, ex);
		}

		logger.info(colExists);

		// retrieve row1
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("finoutc"));
		// scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("height"));
		try {
			ResultScanner resultScanner = table.getScanner(scan);

			for (Result result : resultScanner) {
				// boolean containsColumn =
				// result.containsColumn(Bytes.toBytes("HSE2006"),
				// Bytes.toBytes("AGE"));
				// logger.info("column exists: " + containsColumn);

				// NavigableMap<byte[], NavigableMap<byte[], byte[]>>
				// noVersionMap = result.getNoVersionMap();
				int value = 0;
				try {
					value = Bytes.toInt(result.getRow());
					// byte[] value1 = result.getCellValue().getValue();
					byte[] value1 = result.getCellValue(
							Bytes.toBytes(columnFamily),
							Bytes.toBytes("finoutc")).getValue();
					// byte[] value2 =
					// result.getCellValue(Bytes.toBytes(columnFamily),
					// Bytes.toBytes("height")).getValue();
					// System.out.println(value);
					logger.info(value + " : " + Bytes.toString(value1));
				} catch (Exception e) {
					logger.info("No value for row " + value);
				}
				// logger.info(value + " : " + Bytes.toString(value1) + " : " +
				// Bytes.toString(value2));
			}
		} catch (IOException ex) {
			// System.out.println(ex);
			logger.error(ex);
		}

		// g = new Get(Bytes.toBytes(1));
		// Result r = null;
		// try {
		// r = table.get(g);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		// byte[] bytevalue = r.getValue(Bytes.toBytes("HSE2006"),
		// Bytes.toBytes("age"));
		// System.out.println("age first : " + Bytes.toString(bytevalue));
		// Get g = new Get(Bytes.toBytes(1));
		// Result r = null;
		// try {
		// r = table.get(g);
		// } catch (IOException ex) {
		// logger.error(ex);
		// System.out.println(ex);
		// }
		//
		// byte[] value = r.getValue(Bytes.toBytes("HSE2006"),
		// Bytes.toBytes("age"));
		// logger.info(Bytes.toString(value));
		//
		// System.out.println(new String(tableName));
		// System.out.println(Bytes.toInt(value));
	}

	public void checkAllColumnsInHBaseForDataset(String dataset) {
		HBaseConfiguration config = new HBaseConfiguration();
		HTable table = null;
		try {
			table = new HTable(config, "HSE");
		} catch (IOException ex) {
			logger.error(ex);
			System.out.println(ex);
		}

		LabeledCSVParser labelParser = loadCSVFile(dataset);
		String[] labels = null;
		// get the column labels/headers
		try {
			labels = labelParser.getLabels();
		} catch (IOException e) {
			e.printStackTrace();
		}
		List<String> labelList = Arrays.asList(labels);

		for (String label : labelList) {
			Get g = new Get(Bytes.toBytes(1));
			g.addColumn(Bytes.toBytes(dataset.split("\\.")[0]), Bytes
					.toBytes(label));
			boolean colExists = false;
			try {
				colExists = table.exists(g);
			} catch (IOException ex) {
				System.out.println(ex);
				logger.error(ex);
			}
			// scan.addColumn(Bytes.toBytes(columnFamily),
			// Bytes.toBytes("height"));
			if (colExists) {
				try {
					Scan scan = new Scan();
					scan.addColumn(Bytes.toBytes(dataset.split("\\.")[0]),
							Bytes.toBytes(label));
					ResultScanner resultScanner = table.getScanner(scan);

					int value = 0;
					for (Result result : resultScanner) {
						// boolean containsColumn =
						// result.containsColumn(Bytes.toBytes("HSE2006"),
						// Bytes.toBytes("AGE"));
						// logger.info("column exists: " + containsColumn);

						// NavigableMap<byte[], NavigableMap<byte[], byte[]>>
						// noVersionMap = result.getNoVersionMap();
						try {
							value = Bytes.toInt(result.getRow());
							// byte[] value1 = result.getCellValue().getValue();
							byte[] value1 = result.getCellValue(
									Bytes.toBytes(dataset.split("\\.")[0]),
									Bytes.toBytes(label)).getValue();
							// byte[] value2 =
							// result.getCellValue(Bytes.toBytes(columnFamily),
							// Bytes.toBytes("height")).getValue();
							// System.out.println(value);
							logger.info(value + " : " + Bytes.toString(value1));
							// logger.info(value + " : " +
							// Bytes.toString(value1) + " : " +
							// Bytes.toString(value2));
						} catch (Exception e) {
							System.out.println("Column " + label + " row "
									+ value + " has problem " + e);
						}
					}
					System.out.println(label + " has " + value + " rows");

				} catch (IOException ex) {
					System.out.println("Problem with " + label + " : " + ex);
					java.util.logging.Logger
							.getLogger(CSVUtils.class.getName()).log(
									Level.SEVERE, null, ex);
				}
			} else {
				System.out.println(label + " does not exist");
			}

		}
	}

	private BufferedReader loadFile(String dataset) {
		String directory = getConfig().getString("RTFSearcher.0");
		// stop people from requesting any file
		File rootDirectory = new File(directory);
		String[] list = rootDirectory.list();
		List<String> asList2 = Arrays.asList(list);

		File file = new File(directory, dataset);

		BufferedReader bufferedReader = null;
		try {
			bufferedReader = new BufferedReader(new FileReader(file));
		} catch (FileNotFoundException e1) {
			e1.printStackTrace();
		}

		return bufferedReader;

	}

	private LabeledCSVParser loadCSVFile(String dataset) {
		String directory = getConfig().getString("RawCSVData");
		// stop people from requesting any file
		File rootDirectory = new File(directory);
		String[] list = rootDirectory.list();
		List<String> asList2 = Arrays.asList(list);

		File file = new File(directory, dataset);

		BufferedInputStream bufferedInputStream = null;
		try {
			bufferedInputStream = new BufferedInputStream(new FileInputStream(
					file));
		} catch (FileNotFoundException e1) {
			e1.printStackTrace();
		}

		CSVParser parser = new CSVParser(bufferedInputStream, '\t');

		LabeledCSVParser labelParser = null;
		try {
			labelParser = new LabeledCSVParser(parser);
		} catch (IOException e) {
			e.printStackTrace();
		}

		return labelParser;

	}

	public void sendEmail(String emailAddress, String fileName) {
		logger.info("Sending completion email to: " + emailAddress);
		Configuration emailConfig = null;
		try {
			// If possible, one should try to avoid hard-coding a
			// path in this
			// manner; in a web application, one should place such a
			// file in
			// WEB-INF, and access it using
			// ServletContext.getResourceAsStream.
			// Another alternative is Class.getResourceAsStream.
			// This file contains the javax.mail config properties
			// mentioned above.
			emailConfig = new PropertiesConfiguration("email.properties");
		} catch (Exception e) {
			logger.error("Problem with email properties", e);

		}

		Properties props = new Properties();

		props.put("mail.transport.protocol", emailConfig
				.getString("mail.transport.protocol"));
		props.put("mail.smtps.host", emailConfig.getString("mail.smtps.host"));
		props.put("mail.smtps.auth", "true");

		Session session = Session.getDefaultInstance(props);
		session.setDebug(false);

		MimeMessage message = new MimeMessage(session);
		try {
			// the "from" address may be set in code, or set in the
			// config file under "mail.from" ; here, the latter
			// style is used
			// message.setFrom( new InternetAddress(aFromEmailAddr)
			// );
			// String from = mailServerConfig.getProperty("mail.from");
			// message.setFrom(new InternetAddress(from));
			message.addRecipient(Message.RecipientType.TO, new InternetAddress(
					emailAddress));
			message.setSubject("Archive Ready: " + fileName);
			message.setText("Your archive " + fileName
					+ " is ready for download\nFind it at "
					+ (String) emailConfig.getString("methodbox.url")
					+ "csvarchives" + "\n\nThe MethodBox team");
			String hostName = (String) emailConfig.getString("smtp.host.name");
			int port = emailConfig.getInt("smtp.host.port");
			String user = emailConfig.getString("smtp.auth.user");
			String password = emailConfig.getString("smtp.auth.passwd");

			Transport transport = session.getTransport();
			transport.connect(hostName, port, user, password);
			transport.sendMessage(message, message
					.getRecipients(Message.RecipientType.TO));
			transport.close();
		} catch (MessagingException ex) {
			logger.error("Cannot send email. " + ex);
		}

	}

	public void setConfig(Configuration config) {
		this.config = config;
	}

	public synchronized Configuration getConfig() {
		if (config == null) {
			try {
				config = new PropertiesConfiguration("csv.properties");
			} catch (ConfigurationException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}
		return config;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {

		CSVUtils utils = new CSVUtils();
		// utils.sendEmail("ianwdunlop@googlemail.com", "abcd");
		// eg HSE2006.tab
		utils.checkAllColumnsInHBaseForDataset(args[0]);
		// utils.loadCSVIntoHBase(args[0]);

		// utils.loadCSVColumnIntoHBase("HSE2006.tab", "finoutc");

		// utils.loadCSVIntoHBaseColumnByColumn("HSE2006.tab");
		// eg HSE2006
		// utils.testHBase("HSE2006");

		// utils.readDatasetTest("HSE2006.tab");
		// CSVTester tester = new CSVTester();
		// File file = new File("/Users/Ian/scratch/1991-1992.csv");
		// tester.readFieldNames(file);

		// FileInputStream fstream = null;
		// try {
		// fstream = new FileInputStream(
		//					"/Volumes/Data/obesity_data/2002/HSE2002.tab"); //$NON-NLS-1$ //$NON-NLS-2$
		// } catch (FileNotFoundException e) {
		// e.printStackTrace();
		// }
		// String directory = Messages.getString("RTFSearcher.0");
		// String year = "2002";
		// CSVTester tester = new CSVTester();
		// File file = new File(directory + "/" + year + "/HSE" + year +
		// ".tab");
		// List<String> labelList = new ArrayList<String>();
		// labelList.add("CHILD_WT");
		// tester.readFieldNames(file, "HSE" + year, null);

		// BufferedReader reader = new BufferedReader(new InputStreamReader(
		// fstream));
		// String line;
		// String xmlDoc = null;
		// try {
		// while ((line = reader.readLine()) != null) {
		// xmlDoc = xmlDoc + line;
		// }
		// } catch (IOException e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// }
		// try {
		// JSONObject jsonObject = XML.toJSONObject(xmlDoc);
		// String jsonString = jsonObject.toString();
		// BufferedWriter out = new BufferedWriter(new FileWriter(
		// "/Users/Ian/scratch/json.js"));
		// out.write(jsonString);
		// out.close();
		//
		// } catch (JSONException e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// }
		// // read all the field names
		// // List<String> readFieldNames = tester.readFieldNames(file);
		// // String name = file.getName();
		// // String replace = name.replace(".", "_");
		// // tester.writeTable(readFieldNames, "HSE2002");
		// // File file2 = new
		// File("/Users/Ian/obesity_data/2002/2002-data.tab");
		// // tester.writeSmallTables(file2, readFieldNames, "HSE2002");
		// catch (FileNotFoundException e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// } catch (IOException e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// }

		// file = new File("/Users/Ian/obesity_data/1993/1993-headers.tab");
		// readFieldNames = tester.readFieldNames(file);
		// // name = file.getName();
		// // replace = name.replace(".", "_");
		// tester.writeTable(readFieldNames, "HSE1993");
		//
		// file = new File("/Users/Ian/obesity_data/1994/1994-headers.tab");
		// readFieldNames = tester.readFieldNames(file);
		// // name = file.getName();
		// // replace = name.replace(".", "_");
		// tester.writeTable(readFieldNames, "HSE1994");

	}
}
